﻿
Imports System.Runtime.CompilerServices
Imports System.IO
Imports System.Net.Sockets
Imports osi.root.template
Imports osi.root.constants
Imports osi.root.connector
Imports osi.root.delegates
Imports osi.root.utils
Imports osi.root.formation
Imports osi.root.envs

Public Module _stream
    <Extension()> Public Function write(ByVal s As Stream,
                                        ByVal buff() As Byte,
                                        ByVal offset As Int32,
                                        ByVal count As Int32) As event_comb
        Return create(Function() As Boolean
                          Return Not s Is Nothing AndAlso
                                 s.CanWrite() AndAlso
                                 offset >= 0 AndAlso
                                 count >= 0 AndAlso
                                 array_size(buff) >= offset + count
                      End Function,
                      Function() As event_comb
                          Return New event_comb_async_operation(
                                            Function(a As AsyncCallback) As IAsyncResult
                                                Return s.BeginWrite(buff,
                                                                    offset,
                                                                    count,
                                                                    a,
                                                                    Nothing)
                                            End Function,
                                            Sub(ar As IAsyncResult)
                                                s.EndWrite(ar)
                                            End Sub)
                      End Function)
    End Function

    <Extension()> Public Function write(ByVal s As Stream, ByVal buff() As Byte, ByVal count As Int32) As event_comb
        Return write(s, buff, 0, count)
    End Function

    <Extension()> Public Function write(ByVal s As Stream, ByVal buff() As Byte) As event_comb
        Return write(s, buff, array_size(buff))
    End Function

    'wrongly commented in msdn, it may return 0 even when the connection is still alive.
    <Extension()> Public Function should_retry_when_read_zero(ByVal i As Stream) As Boolean
        Return TypeOf i Is NetworkStream
    End Function

    <Extension()> Public Function read(ByVal s As Stream,
                                       ByVal b() As Byte,
                                       ByVal offset As Int32,
                                       ByVal count As Int32,
                                       ByVal retry_when_read_zero As Boolean,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Dim ec As event_comb = Nothing
        Return New event_comb(Function() As Boolean
                                  If r Is Nothing Then
                                      r = New pointer(Of Int32)()
                                  Else
                                      r.set(0)
                                  End If
                                  ec = create(Function() As Boolean
                                                  Return Not s Is Nothing AndAlso
                                                         s.CanRead() AndAlso
                                                         offset >= 0 AndAlso
                                                         count >= 0 AndAlso
                                                         array_size(b) >= offset + count
                                              End Function,
                                              Function() As event_comb
                                                  Return event_comb_async_operation.ctor(
                                                              Function(ac As AsyncCallback) As IAsyncResult
                                                                  Return s.BeginRead(b,
                                                                                     offset,
                                                                                     count,
                                                                                     ac,
                                                                                     Nothing)
                                                              End Function,
                                                              Function(ar As IAsyncResult) As Int32
                                                                  Return s.EndRead(ar)
                                                              End Function,
                                                              r)
                                              End Function)
                                  Return waitfor(ec) AndAlso
                                         goto_next()
                              End Function,
                              Function() As Boolean
                                  If ec.end_result() Then
                                      If (+r) = 0 AndAlso count > 0 AndAlso retry_when_read_zero Then
                                          Return waitfor(socket_poll_interval_ms) AndAlso
                                                 goto_begin()
                                      Else
                                          Return goto_end()
                                      End If
                                  Else
                                      Return False
                                  End If
                              End Function)
    End Function

    <Extension()> Public Function read(ByVal s As Stream,
                                       ByVal b() As Byte,
                                       ByVal offset As Int32,
                                       ByVal count As Int32,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, b, offset, count, should_retry_when_read_zero(s), r)
    End Function

    <Extension()> Public Function read(Of T As _int64)(
                                       ByVal s As Stream,
                                       ByVal o As qless2_stream(Of Byte, T),
                                       ByVal count As Int64,
                                       ByVal buff_size As Int32,
                                       ByVal retry_when_read_zero As Boolean,
                                       Optional ByVal result As pointer(Of Int32) = Nothing) As event_comb
        Dim b() As Byte = Nothing
        Dim rec As event_comb = Nothing
        Dim wec As event_comb = Nothing
        Dim r As pointer(Of Int32) = Nothing
        Return New event_comb(Function() As Boolean
                                  If o Is Nothing OrElse buff_size <= 0 Then
                                      Return False
                                  Else
                                      ReDim b(buff_size - 1)
                                      r = New pointer(Of Int32)()
                                      Return goto_next()
                                  End If
                              End Function,
                              Function() As Boolean
                                  r.set(0)
                                  assert(count > 0)
                                  rec = read(s, b, CInt(min(buff_size, count)), r)
                                  Return waitfor(rec) AndAlso
                                         goto_next()
                              End Function,
                              Function() As Boolean
                                  If (+r) > 0 Then
                                      assert((+r) <= array_size(b))
                                      If (+r) < array_size(b) Then
                                          Dim tmp() As Byte = Nothing
                                          ReDim tmp((+r) - 1)
                                          memcpy(tmp, b, (+r))
                                          wec = o.push(tmp, npos)
                                      Else
                                          wec = o.push(b, npos)
                                      End If
                                      Return waitfor(wec) AndAlso
                                             goto_next()
                                  Else
                                      Return rec.end_result() AndAlso
                                             goto_end()
                                  End If
                              End Function,
                              Function() As Boolean
                                  If wec.end_result() Then
                                      count -= (+r)
                                      eva(result, (+result) + (+r))
                                      If rec.end_result() Then
                                          If count > 0 Then
                                              Return [goto](1)
                                          Else
                                              Return goto_end()
                                          End If
                                      Else
                                          Return False
                                      End If
                                  Else
                                      Return False
                                  End If
                              End Function)
    End Function

    <Extension()> Public Function read(Of T As _int64)(
                                       ByVal s As Stream,
                                       ByVal o As qless2_stream(Of Byte, T),
                                       ByVal count As Int64,
                                       ByVal buff_size As Int32,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, o, count, buff_size, should_retry_when_read_zero(s), r)
    End Function

    <Extension()> Public Function read(Of T As _int64)(
                                       ByVal s As Stream,
                                       ByVal o As qless2_stream(Of Byte, T),
                                       ByVal count As Int64,
                                       ByVal retry_when_read_zero As Boolean,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, o, count, 4096, retry_when_read_zero, r)
    End Function

    <Extension()> Public Function read(Of T As _int64)(
                                       ByVal s As Stream,
                                       ByVal o As qless2_stream(Of Byte, T),
                                       ByVal count As Int64,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, o, count, should_retry_when_read_zero(s), r)
    End Function

    <Extension()> Public Function read(ByVal s As Stream,
                                       ByVal b() As Byte,
                                       ByVal count As Int32,
                                       ByVal retry_when_read_zero As Boolean,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, b, 0, count, retry_when_read_zero, r)
    End Function

    <Extension()> Public Function read(ByVal s As Stream,
                                       ByVal b() As Byte,
                                       ByVal count As Int32,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, b, count, should_retry_when_read_zero(s), r)
    End Function

    <Extension()> Public Function read(ByVal s As Stream,
                                       ByVal b() As Byte,
                                       ByVal retry_when_read_zero As Boolean,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, b, array_size(b), retry_when_read_zero, r)
    End Function

    <Extension()> Public Function read(ByVal s As Stream,
                                       ByVal b() As Byte,
                                       Optional ByVal r As pointer(Of Int32) = Nothing) As event_comb
        Return read(s, b, should_retry_when_read_zero(s), r)
    End Function

    Private Function convect_one_direction(ByVal i As Stream,
                                           ByVal o As Stream,
                                           ByVal buff_size As Int32,
                                           ByVal idle_time_ms As Int64,
                                           ByVal total_bytes As pointer(Of Int64),
                                           ByVal last_active_time_ms As pointer(Of Int64)) As event_comb
        assert(Not last_active_time_ms Is Nothing)
        Dim ec As event_comb = Nothing
        Dim r As pointer(Of Int32) = Nothing
        Dim always_retry As Boolean = False
        always_retry = should_retry_when_read_zero(i)
        eva(total_bytes, 0)
        Return success_lifetime_event_comb(Function() As Boolean
                                               Return (nowadays.milliseconds() - (+last_active_time_ms)) >= _
                                                       idle_time_ms OrElse
                                                      Not i.CanRead() OrElse
                                                      Not o.CanWrite()
                                           End Function,
                                           Function() As Boolean
                                               If r Is Nothing Then
                                                   r = New pointer(Of Int32)()
                                               Else
                                                   r.set(0)
                                               End If
                                               ec = copy_to_single(i, o, buff_size, r, False)
                                               Return waitfor(ec) AndAlso
                                                      goto_next()
                                           End Function,
                                           Function() As Boolean
                                               If ec.end_result() Then
                                                   If (+r) > 0 Then
                                                       Return eva(last_active_time_ms,
                                                                  nowadays.milliseconds()) AndAlso
                                                              eva(total_bytes, (+total_bytes) + (+r)) AndAlso
                                                              goto_begin()
                                                   ElseIf always_retry Then
                                                       Return waitfor(socket_poll_interval_ms) AndAlso
                                                              goto_begin()
                                                   Else
                                                       Return goto_end()
                                                   End If
                                               Else
                                                   Return goto_end()
                                               End If
                                           End Function)
    End Function

    <Extension()> Public Function convect(ByVal i As Stream,
                                          ByVal o As Stream,
                                          Optional ByVal buff_size As Int32 = 4096,
                                          Optional ByVal idle_time_ms As Int64 = max_int64,
                                          Optional ByVal total_bytes As pointer(Of Int64) = Nothing) As event_comb
        Dim ec_a As event_comb = Nothing
        Dim ec_b As event_comb = Nothing
        Dim ac As pointer(Of Int64) = Nothing
        Dim bc As pointer(Of Int64) = Nothing
        Dim la As pointer(Of Int64) = Nothing
        Return New event_comb(Function() As Boolean
                                  ac = New pointer(Of Int64)()
                                  bc = New pointer(Of Int64)()
                                  la = New pointer(Of Int64)()
                                  eva(la, nowadays.milliseconds())
                                  ec_a = convect_one_direction(i, o, buff_size, idle_time_ms, ac, la)
                                  ec_b = convect_one_direction(o, i, buff_size, idle_time_ms, bc, la)
                                  Return waitfor(ec_a, ec_b) AndAlso
                                         goto_next()
                              End Function,
                              Function() As Boolean
                                  Return eva(total_bytes, (+ac) + (+bc)) AndAlso
                                         ec_a.end_result() AndAlso
                                         (+ac) > 0 AndAlso
                                         ec_b.end_result() AndAlso
                                         (+bc) > 0 AndAlso
                                         goto_end()
                              End Function)
    End Function

    Private Function copy_to_single(ByVal i As Stream,
                                    ByVal o As Stream,
                                    ByVal count As Int32,
                                    ByVal r As pointer(Of Int32),
                                    ByVal retry_when_read_zero As Boolean) As event_comb
        Dim rec As event_comb = Nothing
        Dim wec As event_comb = Nothing
        Dim b() As Byte = Nothing
        Return New event_comb(Function() As Boolean
                                  If count < 0 Then
                                      Return False
                                  ElseIf count = 0 Then
                                      Return goto_end()
                                  Else
                                      ReDim b(count - 1)
                                      If r Is Nothing Then
                                          r = New pointer(Of Int32)()
                                      Else
                                          r.set(0)
                                      End If
                                      rec = read(i, b, retry_when_read_zero, r)
                                      Return waitfor(rec) AndAlso
                                             goto_next()
                                  End If
                              End Function,
                              Function() As Boolean
                                  If rec.end_result() Then
                                      If +r = 0 Then
                                          Return goto_end()
                                      Else
                                          wec = write(o, b, +r)
                                          Return waitfor(wec) AndAlso
                                                 goto_next()
                                      End If
                                  Else
                                      If isdebugmode() Then
                                          raise_error(error_type.warning, "failed to read from input stream")
                                      End If
                                      Return False
                                  End If
                              End Function,
                              Function() As Boolean
                                  If isdebugmode() AndAlso Not wec.end_result() Then
                                      raise_error(error_type.warning, "failed to write to the output stream")
                                  End If
                                  Return wec.end_result() AndAlso
                                         goto_end()
                              End Function)
    End Function

    Private Function unlimited_copy_to(ByVal i As Stream,
                                       ByVal o As Stream,
                                       ByVal buff_size As Int32,
                                       ByVal single_copy_timeout_ms As Int64,
                                       ByVal result As pointer(Of Int64)) As event_comb
        Dim r As pointer(Of Int32) = Nothing
        Dim ec As event_comb = Nothing
        r = New pointer(Of Int32)()
        Dim always_retry As Boolean = False
        always_retry = should_retry_when_read_zero(i)
        If result Is Nothing Then
            result = New pointer(Of Int64)()
        End If
        Return event_comb.repeat(Function(x As Int32) As Boolean
                                     Return ec Is Nothing OrElse (+r) > 0
                                 End Function,
                                 Function() As event_comb
                                     Return New event_comb(Function() As Boolean
                                                               r.set(0)
                                                               ec = copy_to_single(i,
                                                                                   o,
                                                                                   buff_size,
                                                                                   r,
                                                                                   always_retry)
                                                               Return waitfor(ec, single_copy_timeout_ms) AndAlso
                                                                      goto_next()
                                                           End Function,
                                                           Function() As Boolean
                                                               If ec.end_result() AndAlso (+r) > 0 Then
                                                                   Return eva(result, (+result) + (+r)) AndAlso
                                                                          goto_end()
                                                               Else
                                                                   'not ec.end_result() orelse (+r) = 0
                                                                   r.set(0)
                                                                   Return (+result) > 0 AndAlso
                                                                          goto_end()
                                                               End If
                                                           End Function)
                                 End Function)
    End Function

    Private Function limited_copy_to(ByVal i As Stream,
                                     ByVal o As Stream,
                                     ByVal count As Int64,
                                     ByVal buff_size As Int32,
                                     ByVal single_copy_timeout_ms As Int64,
                                     ByVal result As pointer(Of Int64)) As event_comb
        assert(count > 0)
        Dim r As pointer(Of Int32) = Nothing
        Dim ec As event_comb = Nothing
        r = New pointer(Of Int32)()
        Dim always_retry As Boolean = False
        always_retry = should_retry_when_read_zero(i)
        Return event_comb.repeat(Function(x As Int32) As Boolean
                                     Return count > 0 AndAlso (ec Is Nothing OrElse (+r) > 0)
                                 End Function,
                                 Function() As event_comb
                                     Return New event_comb(Function() As Boolean
                                                               ec = copy_to_single(i,
                                                                                   o,
                                                                                   min(buff_size, count),
                                                                                   r,
                                                                                   always_retry)
                                                               Return waitfor(ec, single_copy_timeout_ms) AndAlso
                                                                      goto_next()
                                                           End Function,
                                                           Function() As Boolean
                                                               If ec.end_result() AndAlso (+r) > 0 Then
                                                                   count -= (+r)
                                                                   Return eva(result, (+result) + (+r)) AndAlso
                                                                          goto_end()
                                                               Else
                                                                   'not ec.end_result() orelse (+r) = 0
                                                                   r.set(0)
                                                                   Return ec.end_result() AndAlso
                                                                          goto_end()
                                                               End If
                                                           End Function)
                                 End Function)
    End Function

    <Extension()> Public Function copy_to(ByVal i As Stream,
                                          ByVal o As Stream,
                                          ByVal count As Int64,
                                          ByVal buff_size As Int32,
                                          ByVal single_copy_timeout_ms As Int64,
                                          Optional ByVal result As pointer(Of Int64) = Nothing) As event_comb
        If count > 0 Then
            Return limited_copy_to(i, o, count, buff_size, single_copy_timeout_ms, result)
        ElseIf count = 0 Then
            Return New event_comb(Function() As Boolean
                                      Return eva(result, 0) AndAlso
                                             goto_end()
                                  End Function)
        Else
            Return unlimited_copy_to(i, o, buff_size, single_copy_timeout_ms, result)
        End If
    End Function

    <Extension()> Public Function copy_to(ByVal i As Stream,
                                          ByVal o As Stream,
                                          ByVal count As Int64,
                                          ByVal single_copy_timeout_ms As Int64,
                                          Optional ByVal result As pointer(Of Int64) = Nothing) As event_comb
        Return copy_to(i, o, count, 4096, single_copy_timeout_ms, result)
    End Function

    <Extension()> Public Function copy_to(ByVal i As Stream,
                                          ByVal o As Stream,
                                          ByVal single_copy_timeout_ms As Int64,
                                          Optional ByVal result As pointer(Of Int64) = Nothing) As event_comb
        Return copy_to(i, o, npos, single_copy_timeout_ms, result)
    End Function

    <Extension()> Public Function copy_to(ByVal i As Stream,
                                          ByVal o As Stream,
                                          Optional ByVal result As pointer(Of Int64) = Nothing) As event_comb
        Return copy_to(i, o, max_int64, result)
    End Function

    Private ReadOnly empty_buff() As Byte

    Sub New()
        '-1 means nothing?
        ReDim empty_buff(0)
        assert(npos < 0)
    End Sub

    'not only whether the stream can write, but also try to write an empty buff
    <Extension()> Public Function can_read(ByVal i As Stream) As Boolean
        Return Not i Is Nothing AndAlso i.CanRead()
#If 0 Then
        If i Is Nothing OrElse Not i.CanRead() Then
            Return False
        Else
            Try
                i.read(empty_buff, 0, 0)
                Return True
            Catch
                Return False
            End Try
        End If
#End If
    End Function

    <Extension()> Public Function can_write(ByVal i As Stream) As Boolean
        Return Not i Is Nothing AndAlso i.CanWrite()
#If 0 Then
        If i Is Nothing OrElse Not i.CanWrite() Then
            Return False
        Else
            Try
                i.write(empty_buff, 0, 0)
                Return True
            Catch
                Return False
            End Try
        End If
#End If
    End Function

    <Extension()> Public Function data_available(ByVal i As NetworkStream) As ternary
        If i Is Nothing Then
            Return ternary.unknown
        Else
            Try
                If i.DataAvailable() Then
                    Return ternary.true
                Else
                    Return ternary.false
                End If
            Catch
                Return ternary.unknown
            End Try
        End If
    End Function
End Module
